/*
 * MIT License
 *
 * Copyright (c) 2020 wen.gu <454727014@qq.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

/***************************************************************************
 * Name: message_worker.h
 *
 * Purpose: Define and implementation a worker for process message
 *
 * Developer:
 *   wen.gu , 2023-08-01
 *
 * TODO:
 *
 ***************************************************************************/

#ifndef __XBUS_TRANSPORT_MESSAGE_WORKER_H__
#define __XBUS_TRANSPORT_MESSAGE_WORKER_H__

#include <string>
#include <memory>
#include <functional>

#include "xbus/core/xtype.h"
#include "xbus/core/xerror.h"
#include "xbus/osal/xthread.h"

namespace xbus::transport {

template<typename MsgQueue>
class MessageWorker {
public:
    using XErrC = xbus::core::XErrC;
    using XThreadPtr = osal::XThread::XThreadPtr;
    using MsgQueuePtr = std::shared_ptr<MsgQueue>;
    using WorkerHandler = std::function<void(core::XComId com_id, MsgQueuePtr mq)>;
public:
    MessageWorker(const std::string& name, int32_t priority)
    :worker_name_(name),
    priority_(priority) {}
    virtual ~MessageWorker() { stop(); }  
public:
    virtual void set_msg_queue(MsgQueuePtr mq_ptr) { msg_queue_ = mq_ptr; }
    virtual MsgQueuePtr msg_queue() const { return msg_queue_; }

    /**if max size == 0, then indicat unlimited the size of message queue */
    virtual XErrC alloc_msg_queue(uint32_t max_size = 0) {
        if (msg_queue_ != nullptr) {
            return XErrC::InvalidStatus;
        }

        msg_queue_ = std::make_shared<MsgQueue>(max_size);
        return msg_queue_ ? XErrC::OK : XErrC::Undefined;
    } 

    /** only for the io which has send/recv thread/worker */
    virtual int32_t priority() { return priority_; }
    virtual void set_com_id(core::XComId com_id) { com_id_ = com_id; }
    virtual core::XComId com_id() { return com_id_; }
    
    virtual void set_worker_handler(WorkerHandler&& hanlder) { worker_handler_ = hanlder; }

    virtual const std::string& name() { return worker_name_; }
public:
    virtual XErrC start() { // start working thread
        if (worker_thread_) {
            return XErrC::InvalidStatus;
        }

        worker_thread_ = std::make_shared<osal::XThread>(worker_name_, std::bind(&MessageWorker::do_worker, this));

        is_running_ = true;    
        msg_queue_->start();

        worker_thread_->start(); //todo is worker thread start fialed?

        return XErrC::OK;
    }

    virtual XErrC stop() {  //stop will call clear() also 
        //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
        if (worker_thread_ == nullptr) {
            return XErrC::InvalidStatus;
        }
        //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
        is_running_ = false;
        if (worker_thread_) {
            worker_thread_->join();
            worker_thread_ = nullptr;
        }

        //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
        clear();
        //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
        return msg_queue_->stop();
    }

    virtual XErrC clear(){ // clear(drop) all data cached message in worker.
        if (msg_queue_) { msg_queue_->clear(); }
        return XErrC::OK;
    }

protected:
    virtual void do_worker() {
        while (is_running_) {
            if (worker_handler_) {
                worker_handler_(com_id_, msg_queue_);
            } else {
            osal::SleepMs(10); 
            }
        }
    }

protected:
    bool is_running_ = false;
    core::XComId com_id_ = INVALID_XCOM_ID;
    int32_t priority_;
    MsgQueuePtr msg_queue_ = nullptr;
    XThreadPtr worker_thread_;
    std::string worker_name_;
    WorkerHandler worker_handler_;    
};




} //namespace xbus::transport


#endif /** !__XBUS_TRANSPORT_MESSAGE_WORKER_H__ */